package com.shujia.flink.core;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo02WordCountSimple {
    public static void main(String[] args) throws Exception {
        // 创建Flink环境
        // 不需要设置AppName以及Master运行模式，Flink自动识别运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Source：通过Socket模拟流式环境，做到数据源源不断 nc -lk 8888
        // 数据会对CPU的线程以轮询的方式接收过来
        DataStream<String> socketDS = env.socketTextStream("master", 8888);

        // Transformation
        // 第一个String表示上游的DS中每一条数据的类型，第二个String表示flatMap输出的每一条数据的类型
        // 第一种方式：使用匿名内部类
        DataStream<String> wordsDS = socketDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            /**
             * 第一个参数：输入进来的每一条数据
             * 第二个参数：将数据发送到下游（输出）
             */
            public void flatMap(String line, Collector<String> out) throws Exception {
                // 将每一条数据的每一个单词切分出来 a,a,a,a,a
                for (String word : line.split(",")) {
                    // 通过out将每个单词发送到下游
                    out.collect(word);
                }
            }
        });

        // 第二种方式：使用自定义类实现接口
        socketDS.flatMap(new MyFlatMapFunction()).print();

        // 第三种方式：使用匿名函数（lambda表达式）
        socketDS.flatMap((line,out)->{
            for (String word : line.split(",")) {
                out.collect(word);
            }
        },Types.STRING).print();


        // 将每个单词转换成KV格式，以单词本身作为K，1作为V
        DataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING,Types.INT));

        // 将每个单词进行分组，完成最终的Count统计

        // 按照DS中的每一条数据的某一部分进行分组
        KeyedStream<Tuple2<String, Integer>, String> keyedDS = kvDS.keyBy(kv -> kv.f0);// keyBy之后的数据流会按Key以一定的策略将数据发送到不同的线程中，相同的Key的数据会进入同一个线程

        // 汇总
        // Flink中的计算默认是有状态的：每一次的结果更新会参考上一次的状态(结果)
        DataStream<Tuple2<String, Integer>> wordCntDS = keyedDS.sum(1);// 1 表示按照 二元组的第二列进行sum统计

        // Sink
//        socketDS.print();
//        wordsDS.print();
//        kvDS.print();
//        keyedDS.print();
//        wordCntDS.print();
        // 启动任务
        env.execute();

    }
}

// 第二种方式：使用自定义类实现接口
class MyFlatMapFunction implements FlatMapFunction<String,String>{

    @Override
    public void flatMap(String line, Collector<String> out) throws Exception {
        for (String word : line.split(",")) {
            out.collect(word);
        }
    }
}
